package com.ark.cloud.java.persistence.listener;



import com.alibaba.fastjson.JSONArray;
import com.ark.cloud.java.persistence.entity.*;

import com.ark.cloud.java.persistence.service.*;
import com.ark.cloud.java.persistence.utils.ByteUtils;
import com.ark.cloud.java.persistence.utils.KafkaAnalysisUtils;
import com.ark.cloud.java.persistence.utils.UUIDUtil;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.UUID;

@Component
@Slf4j

public class DataListener {

    public static final int HEAD_SIZE = 20;

    @Autowired
    private SensorDataService sensorDataService;
    @Autowired
    private PlatStateService platStateService;
    @Autowired
    private FireEventService fireEventService;
    @Autowired
    private ExplodeEventService explodeEventService;
    @Autowired
    private DetectEventService detectEventService;
    @Autowired
    private BGCMDService bgcmdService;
    @Autowired
    private  SensorOnOffEventService sensorOnOffEventService;
    @Autowired
    private CommunicationEventService communicationEventService;
    @Autowired
    private  DestoryEventService destoryEventService;
    @Autowired
    private  ScenarioDataService scenarioDataService;
    @Autowired
    private ContactStateService contactStateService;
    @Autowired
    private WaypointCmdService waypointCmdService;
    @Autowired
    private  SimEngineStateService simEngineStateService;
    @Autowired
    private  AircraftStatusService aircraftStatusService;
    @Autowired
    private ReplyCmdService replyCmdService;
    @Autowired
    private AircraftLaunchCmdService aircraftLaunchCmdService;
    @Autowired
    private SpeedCmdService speedCmdService;
    @Autowired
    private JammerCmdService jammerCmdService;
    @Autowired
    private GuideCmdService guideCmdService;
    @Autowired
    private LuaCmdService luaCmdService;
    @Autowired
    private LuaCmdReplyService luaCmdReplyService;

    @Autowired
    private ScenarioStateService scenarioStateService;

    @Autowired
    private JammerEventService jammerEventService;

    @Autowired
    private ReconnoiterCmdSerivce reconnoiterCmdSerivce;

    @Autowired
    private FireJGEventService fireJGEventService;
    @Autowired
    private FireDNEventService fireDNEventService;
    @Autowired
    private FireCKEventService fireCKEventService;
    @Autowired
    private EndGameJGEventService endGameJGEventService;
    @Autowired
    private EndGameDNEventService endGameDNEventService;
    @Resource
    private EndGameCKEventService endGameCKEventService;

    @Autowired
    private FireJGCmdService fireJGCmdService;
    @Autowired
    private FireDNCmdService fireDNCmdService;
    @Autowired
    private FireCKCmdService fireCKCmdService;
    @Autowired
    private SatelliteCmdService satelliteCmdService;

    @Autowired
    private SatOrbitService satOrbitService;



    //平台数据
    @KafkaListener(topics = "Topic_PlatState")
    public void onMessagePlat(byte[] bytes){
//        Thread thread = Thread.currentThread();
//        log.info("thread1111: {}", thread.getName());

        System.out.println("bytes.length:"+bytes.length);
        byte[] pd = ByteUtils.splitArray(bytes, 0, HEAD_SIZE);
        MSGHead msgHead = KafkaAnalysisUtils.analysisMsgHead(bytes);
        System.out.println(msgHead);

        int strNum = KafkaAnalysisUtils.analysisMsgHeadStrNum(pd);

        System.out.println("strNum:"+strNum);

        List<PlatState> platStates = new ArrayList<>();

        for (int i = 0; i < strNum; i++) {

            byte[] bytes2 = ByteUtils.splitArray(bytes, i * 252 + HEAD_SIZE, 252);

            PlatState platState = KafkaAnalysisUtils.analysisPlatState(bytes2);
            System.out.println("platState:"+platState);
            platStates.add(platState);
        }

         System.out.println("platStates = " + platStates);
         log.info("platStates: {}", platStates.toString());
         for(PlatState platState : platStates){

             platStateService.savePlatStateInfo(platState);
         }
    }

    //传感器数据  SensorData
//    分割的起点和长度之和必须小于数组的长度！
     @KafkaListener(topics = "Topic_SensorData")
    public void onMessageSensorData(byte[] bytes){

//        Thread thread = Thread.currentThread();
//        log.info("thread2222: {}", thread.getName());
         System.out.println("传感器数据Topic_SensorData = " + bytes);

        byte[] pd = ByteUtils.splitArray(bytes, 0, HEAD_SIZE);
        MSGHead msgHead = KafkaAnalysisUtils.analysisMsgHead(bytes);
        int strNum = KafkaAnalysisUtils.analysisMsgHeadStrNum(pd);

        List<SensorData> SensorDataList = new ArrayList<>();
        int sumLength = 0;

        for (int i = 0; i < strNum; i++) {

            byte[] bytes2 = ByteUtils.splitArray(bytes, i * 170 + HEAD_SIZE + sumLength, 170);
            SensorData sensorData = KafkaAnalysisUtils.analysisSensorData(bytes2);

            short number = sensorData.getEmitterNumber();
            ArrayList<Beam> beams = new ArrayList<>();
            for (int j = 0; j < number; j++) {

                byte[] beamsBytes = ByteUtils.splitArray(bytes, (i + 1) * 170 + HEAD_SIZE + sumLength, 24);
                Beam beam = KafkaAnalysisUtils.analysisBeams(beamsBytes);
                beams.add(beam);
                sumLength += 24;
            }
            sensorData.setBeams(beams);

            SensorDataList.add(sensorData);
        }

        for(SensorData sensorData : SensorDataList){
            System.out.println("sensorData:"+sensorData);
            log.info("sensorData: {}", sensorData.toString());

            sensorDataService.saveSensorDataInfo(sensorData);
        }

    }
    //3  时统数据 ScenarioState
    @KafkaListener(topics = "Topic_ScenarioState")
    public void onMessageScenarioState(byte[] bytes){
//        Thread thread = Thread.currentThread();
//        log.info("thread3333: {}", thread.getName());

        byte[] bytes2 = ByteUtils.splitArray(bytes, HEAD_SIZE, bytes.length - HEAD_SIZE);


        ScenarioState scenarioState = new ScenarioState();
        scenarioState.setTimestamp(KafkaAnalysisUtils.dateTOString(bytes2) );
        scenarioState.setId(UUIDUtil.next());
        scenarioStateService.save(scenarioState);
    }



    // 4 目标数据 ContactState
    @KafkaListener(topics = "Topic_ContactState")
//    分割的起点和长度之和必须小于数组的长度！
    public void onMessageContact(byte[] bytes){
        System.out.println("目标数据 Topic_ContactState = " + bytes);

        byte[] pd = ByteUtils.splitArray(bytes, 0, 20);
        //得到头文件信息中的strNum
        int strNum = KafkaAnalysisUtils.analysisMsgHeadStrNum(pd);

        System.out.println("strNum = " + strNum);
        int sumLength = 0;
        ArrayList<ContactState> contactStateList = new ArrayList<>();

        try {
            System.out.println("bytes = " + new String(bytes,"UTF-8") + " byte length = " + bytes.length);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        System.out.println("byte length = " + bytes.length);

        for (int i = 0; i < strNum; i++) {

            System.out.println("--------- start ----------");
            byte[] bytes2 = ByteUtils.splitArray(bytes, i * 274 + HEAD_SIZE +sumLength, 274);
            ContactState contactState = KafkaAnalysisUtils.analysisContactState(bytes2);

            Integer recordCount = contactState.getRecordCount();
            System.out.println("recordCount = " + recordCount);
            if (recordCount != 0) {
                System.out.println("asdasd = " + recordCount);

            }
            ArrayList<DetectingRecord> recordEntityList = new ArrayList<>();
            int length = 0;
            for (int j = 0; j < recordCount; j++) {
                byte[] recordBytes = ByteUtils.splitArray(bytes, (i + 1) * 274 + HEAD_SIZE + sumLength + (j * 28) + length, 28);
                DetectingRecord detectingRecord = KafkaAnalysisUtils.analysisRecordList(recordBytes);

                int sensorCount = detectingRecord.getSensorCount();
                System.out.println("sensor count ===== " + sensorCount);

                ArrayList<SensorInfo> entities = new ArrayList<>();

                for (int k = 0; k < sensorCount; k++) {
                    byte[] sensorByte = ByteUtils.splitArray(bytes, (i + 1) * 274 + HEAD_SIZE + sumLength + ((j + 1) * 28) + (k * 43), 43);
//                    byte[] sensorByte = ByteUtils.splitArray(bytes, (i + 1) * 274 + HEAD_SIZE + sumLength + ((j + 1) * 28) + (k * 43), realLegth);
                    SensorInfo sensorEntity = KafkaAnalysisUtils.analysisSensorList(sensorByte);
                    entities.add(sensorEntity);

                }
                length += sensorCount * 43;
                detectingRecord.setSensorList(entities);
                recordEntityList.add(detectingRecord);

            }
            sumLength += recordCount * 28 + length;
            contactState.setRecordList( recordEntityList);

            contactStateList.add(contactState);


        }

        for(ContactState contactState :contactStateList){
            System.out.println("contactState:: " + contactState);

            contactStateService.saveContactStateInfo(contactState);
        }
    }



    //5  卫星轨道数据  SatOrbit
     @KafkaListener(topics = "Topic_SatOrbit")
    public void onMessageSatOrbit(byte[] bytes){
        byte[] pd = ByteUtils.splitArray(bytes, 0, HEAD_SIZE);
        int strNum = KafkaAnalysisUtils.analysisMsgHeadStrNum(pd);

        ArrayList<SatOrbit> satOrbits = new ArrayList<>();
        for (int i = 0; i < strNum; i++) {

            byte[] bytes2 = ByteUtils.splitArray(bytes, i * 222 + HEAD_SIZE, 222);
            SatOrbit satOrbit = KafkaAnalysisUtils.analysisSatOrbit(bytes2);
            satOrbits.add(satOrbit);
//            System.out.println(satOrbit);
        }
         for(SatOrbit satOrbit :satOrbits){
             System.out.println("satOrbit:: " + satOrbit);

             satOrbitService.saveSatOrbit(satOrbit);
         }


    }



//    //   事件接收
     @KafkaListener(topics = "Topic_Event")
    public void onMessageJGGame(byte[] bytes){
         System.out.println("bytes.length:"+bytes.length);
        byte[] pd = ByteUtils.splitArray(bytes, 0, HEAD_SIZE);
        MSGHead msgHead = KafkaAnalysisUtils.analysisMsgHead(pd);
        Integer index = msgHead.getIndex();


        switch (index){
//            E_FireEvent
            case 5:
                byte[] fireBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 147);
                FireEvent fireEvent = KafkaAnalysisUtils.analysisFireEvent(fireBytes);

                System.out.println("fireEvent:;"+fireEvent);
                fireEventService.saveFireEventInfo(fireEvent);

                break;
         //E_ExplodeEvent
            case 6:
                byte[] explodeEventBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 157);
                ExplodeEvent explodeEvent = KafkaAnalysisUtils.analysisExplodeEvent(explodeEventBytes);

                System.out.println("explodeEvent:;"+explodeEvent);
                explodeEventService.saveExplodeEventInfo(explodeEvent);
                break;
            //E_DetectEvent
            case 7:
                byte[] detectEventBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 96);
                DetectEvent detectEvent = KafkaAnalysisUtils.analysisDetectEvent(detectEventBytes);

                System.out.println("detectEvent:;"+detectEvent);
                detectEventService.saveDetectEventInfo(detectEvent);
                break;
            //E_SensorOnOffEvent
            case 9:
                byte[] sensorOnOffEventBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 173);
                SensorOnOffEvent sensorOnOffEvent = KafkaAnalysisUtils.analysisSensorOnOffEvent(sensorOnOffEventBytes);

                System.out.println("sensorOnOffEvent:;"+sensorOnOffEvent);
                sensorOnOffEventService.saveSensorOnOffEventInfo(sensorOnOffEvent);
                break;
            //E_CommunicationEvent
            case 10:
                byte[] communicationEventBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 101);
                CommunicationEvent communicationEvent = KafkaAnalysisUtils.analysisCommunicationEvent(communicationEventBytes);

                System.out.println("communicationEvent:;"+communicationEvent);
                communicationEventService.saveCommunicationEventInfo(communicationEvent);
                break;
            //E_DestoryEvent
            case 11:
                byte[] destoryEventBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 59);
                DestoryEvent destoryEvent = KafkaAnalysisUtils.analysisDestoryEvent(destoryEventBytes);

                System.out.println("destoryEvent:;"+destoryEvent);
                destoryEventService.saveDestoryEventInfo(destoryEvent);
                break;

            //E_JammerEvent  ###
            case 12:
                byte[] jammerEventBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 66);
                JammerEvent jammerEvent = KafkaAnalysisUtils.analysisJammerEvent(jammerEventBytes);

                Integer platNum = jammerEvent.getPlatNum();
                System.out.println("platNum:"+platNum);
                ArrayList<String> jammerEntIdList = new ArrayList<>();
                for (int i = 0; i < platNum; i++) {

                    byte[] aBytes = ByteUtils.splitArray(bytes, i * 37 +66+ HEAD_SIZE , 37);
                    String a = KafkaAnalysisUtils.analysisEntityId(aBytes);
                    System.out.println("a:"+a);
                    jammerEntIdList.add(a);
                }

                jammerEvent.setJammerEntIdList(jammerEntIdList);

                System.out.println("jammerEvent:;"+jammerEvent);
                jammerEventService.saveJammerEvent(jammerEvent);
                break;

            // SimEngineState 引擎状态
            case 13:
                byte[] engineBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 58);
                SimEngineState simEngineState = KafkaAnalysisUtils.analysisSimEngineState(engineBytes);
                System.out.println("simEngineState:;"+simEngineState);
                simEngineStateService.saveSimEngineStateInfo(simEngineState);

                break;

//            AircraftStatus = 14
            case 14:
                byte[] aircraftStatusBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 63);
                AircraftStatus aircraftStatus = KafkaAnalysisUtils.analysisAircraftStatus(aircraftStatusBytes);
                System.out.println("aircraftStatus:;"+aircraftStatus);
                aircraftStatusService.saveAircraftStatusInfo(aircraftStatus);

                break;

//                E_FireJGEvent
            case 32:
                byte[] fireJGEventBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 237);
                FireJGEvent fireJGEvent = KafkaAnalysisUtils.analysisFireJGEvent(fireJGEventBytes);
                System.out.println("fireJGEvent:;"+fireJGEvent);
                fireJGEventService.saveFireJGEvent(fireJGEvent);

                break;
            //      E_FireDNEvent
            case 33:
                byte[] fireDNEventBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 223);
                FireDNEvent fireDNEvent = KafkaAnalysisUtils.analysisFireDNEvent(fireDNEventBytes);
                System.out.println("fireDNEvent:;"+fireDNEvent);
                fireDNEventService.saveFireDNEvent(fireDNEvent);

                break;
            //      E_FireCKEvent
            case 34:
                byte[] fireCKEventBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 221);
                FireCKEvent fireCKEvent = KafkaAnalysisUtils.analysisFireCKEvent(fireCKEventBytes);
                System.out.println("fireCKEvent:;"+fireCKEvent);
                fireCKEventService.saveFireCKEvent(fireCKEvent);

                break;

            //E_EndGameJGEvent
            case 35:
                byte[] endGameJGEventBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 239);
                EndGameJGEvent endGameJGEvent = KafkaAnalysisUtils.analysisEndGameJGEvent(endGameJGEventBytes);
                System.out.println("endGameJGEvent:;"+endGameJGEvent);
                endGameJGEventService.saveEndGameJGEvent(endGameJGEvent);
                break;

//              E_EndGameDNEvent
            case 36:
                byte[] bytes2 = ByteUtils.splitArray(bytes, HEAD_SIZE, 239);
                EndGameDNEvent endGameDNEvent = KafkaAnalysisUtils.analysisEndGameDNEvent(bytes2);
                System.out.println("endGameDNEvent:;"+endGameDNEvent);
                endGameDNEventService.saveEndGameDNEvent(endGameDNEvent);
                break;

//                E_EndGameCKEvent
            case 37:
                //CK 命中/错过事件
                byte[] endGameCKEventBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 223);
                EndGameCKEvent endGameCKEvent = KafkaAnalysisUtils.analysisEndGameCKEvent(endGameCKEventBytes);
                System.out.println("endGameCKEvent:;"+endGameCKEvent);
                endGameCKEventService.saveEndGameCKEvent(endGameCKEvent);
                break;

            default:
                // logger.info("----------接收到未处理Event {} ------------",index);
                break;
        }

    }



    //    //  命令接收
    @KafkaListener(topics = "Topic_Cmd")
    public void onMessageCmd(byte[] bytes) {

        byte[] pd = ByteUtils.splitArray(bytes, 0, HEAD_SIZE);
        MSGHead msgHead = KafkaAnalysisUtils.analysisMsgHead(pd);
        Integer index = msgHead.getIndex();

        switch (index) {
            //        E_ReplyCmd
            case 15:
                byte[] replyCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 59);
                ReplyCmd replyCmd = KafkaAnalysisUtils.analysisReplyCmd(replyCmdBytes);

                System.out.println("replyCmd:;" + replyCmd);
                replyCmdService.saveReplyCmdInfo(replyCmd);
                break;
//
////            AircraftLaunchCmd = 16
            case 16:
                byte[] aircraftLaunchCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 134);
                AircraftLaunchCmd aircraftLaunchCmd = KafkaAnalysisUtils.analysisAircraftLaunchCmd(aircraftLaunchCmdBytes);

                System.out.println("aircraftLaunchCmd:;" + aircraftLaunchCmd);
                aircraftLaunchCmdService.saveAircraftLaunchCmdInfo(aircraftLaunchCmd);
                break;
//                  SpeedCmd = 17
            case 17:
                byte[] speedCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 100);
                SpeedCmd speedCmd = KafkaAnalysisUtils.analysisSpeedCmd(speedCmdBytes);

                System.out.println("speedCmd:;" + speedCmd);
                speedCmdService.saveSpeedCmdInfo(speedCmd);
                break;

//            WaypointCmd = 18
            case 18:
                byte[] waypointCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 98);
                WaypointCmd waypointCmd = KafkaAnalysisUtils.analysisWaypointCmd(waypointCmdBytes);

                short count = waypointCmd.getCount();
                System.out.println("count:"+count);
                ArrayList<Vec3D> posList = new ArrayList<>();
                for (int i = 0; i < count; i++) {

                    byte[] aBytes = ByteUtils.splitArray(bytes, i * 24 +98+ HEAD_SIZE , 24);
                    Vec3D vec3D = KafkaAnalysisUtils.analysisVec3D(aBytes);
                    System.out.println("vec3D:"+vec3D);
                    posList.add(vec3D);
                }

                waypointCmd.setPosList(posList);

                System.out.println("waypointCmd:;" + waypointCmd);
                waypointCmdService.saveWaypointCmdInfo(waypointCmd);

                break;
//            JammerCmd = 19
            case 19:
                byte[] jammerCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 201);
                JammerCmd jammerCmd = KafkaAnalysisUtils.analysisJammerCmd(jammerCmdBytes);

                System.out.println("jammerCmd:;" + jammerCmd);
                jammerCmdService.saveJammerCmdInfo(jammerCmd);
                break;
//            GuideCmd = 20
            case 20:
                byte[]  guideCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 181);
                GuideCmd guideCmd = KafkaAnalysisUtils.analysisGuideCmd(guideCmdBytes);

                System.out.println("guideCmd:;" + guideCmd);
                guideCmdService.saveGuideCmdInfo(guideCmd);
                break;

//            LuaCmd = 21
            case 21:
                byte[]  luaCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 91);
                LuaCmd luaCmd = KafkaAnalysisUtils.analysisLuaCmd(luaCmdBytes);

                System.out.println("luaCmd:;" + luaCmd);
                luaCmdService.saveLuaCmdInfo(luaCmd);
                break;

//            LuaCmdReply = 22
            case 22:
                byte[]  luaCmdReplyBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 61);
                LuaCmdReply luaCmdReply = KafkaAnalysisUtils.analysisLuaCmdReply(luaCmdReplyBytes);

                System.out.println("luaCmdReply:;" + luaCmdReply);
                luaCmdReplyService.saveLuaCmdReplyInfo(luaCmdReply);
                break;

//            ReconnoiterCmd = 23
            case 23:
                byte[]  feconnoiterCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 182);
                ReconnoiterCmd reconnoiterCmd = KafkaAnalysisUtils.analysisReconnoiterCmd(feconnoiterCmdBytes);

                short count2 = reconnoiterCmd.getCount();
                System.out.println("count2:"+count2);

                ArrayList<Vec3D> posList2 = new ArrayList<>();
                for (int i = 0; i < count2; i++) {

                    byte[] aBytes = ByteUtils.splitArray(bytes, i * 24 +182+ HEAD_SIZE , 24);
                    Vec3D vec3D = KafkaAnalysisUtils.analysisVec3D(aBytes);
                    System.out.println("vec3D:"+vec3D);
                    posList2.add(vec3D);
                }

                reconnoiterCmd.setAreaList(posList2);

                System.out.println("reconnoiterCmd:;" + reconnoiterCmd);
                reconnoiterCmdSerivce.saveReconnoiterCmd(reconnoiterCmd);
                break;

//            FireJGCmd = 28
            case 28:
                byte[]  fireJGCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 214);
                FireJGCmd fireJGCmd = KafkaAnalysisUtils.analysisFireJGCmd(fireJGCmdBytes);

                System.out.println("fireJGCmd:;" + fireJGCmd);
                fireJGCmdService.saveFireJGCmd(fireJGCmd);
                break;
//            FireDNCmd = 29
            case 29:
                byte[]  fireDNCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 228);
                FireDNCmd fireDNCmd = KafkaAnalysisUtils.analysisFireDNCmd(fireDNCmdBytes);

                System.out.println("fireDNCmd:;" + fireDNCmd);
                fireDNCmdService.saveFireDNCmd(fireDNCmd);
                break;

//            FireCKCmd = 30
            case 30:
                byte[]  fireCKCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 176);
                FireCKCmd fireCKCmd = KafkaAnalysisUtils.analysisFireCKCmd(fireCKCmdBytes);

                System.out.println("fireCKCmd:;" + fireCKCmd);
                fireCKCmdService.saveFireCKCmd(fireCKCmd);
                break;
//            SatelliteCmd = 31
            case 31:
                byte[]  satelliteCmdBytes = ByteUtils.splitArray(bytes, HEAD_SIZE, 937);
                SatelliteCmd satelliteCmd = KafkaAnalysisUtils.analysisSatelliteCmd(satelliteCmdBytes);

                System.out.println("satelliteCmd:;" + satelliteCmd);
                satelliteCmdService.saveSatelliteCmd(satelliteCmd);

                break;

            default:
                break;
        }



    }


//   想定 ScenarioData
    @KafkaListener(topics = "Topic_Scenario")
    public void onMessageScenarioData(byte[] bytes){

        byte[] pd = ByteUtils.splitArray(bytes, 0, HEAD_SIZE);
        int strNum = KafkaAnalysisUtils.analysisMsgHeadStrNum(pd);
        ArrayList<ScenarioData> scenarioDataArrayList = new ArrayList<>();
        for (int i = 0; i < strNum; i++) {

            byte[] bytes2 = ByteUtils.splitArray(bytes, i * 28 + HEAD_SIZE, 28);
            ScenarioData scenarioData = KafkaAnalysisUtils.analysisScenarioData(bytes2);
            scenarioDataArrayList.add(scenarioData);

        }
        for(ScenarioData scenarioData :scenarioDataArrayList){
            System.out.println("scenarioData:: " + scenarioData);

            scenarioDataService.saveScenarioDataInfo(scenarioData);
        }

    }

//    变轨事件(BGCMD)
    @KafkaListener(topics = "Topic_BGCMD")
    public void onMessageBGCMD(byte[] bytes){

        byte[] pd = ByteUtils.splitArray(bytes, 0, HEAD_SIZE);
        int strNum = KafkaAnalysisUtils.analysisMsgHeadStrNum(pd);
        ArrayList<BGCMD> bgcmds = new ArrayList<>();
        int sumLength = 0;

        for (int i = 0; i < strNum; i++) {
            BGCMD bgcmd = new BGCMD();
            byte[] bGCMDHEADArray = ByteUtils.splitArray(bytes, i * 166 + HEAD_SIZE + sumLength, 166);

            BGCMDHEAD bgcmdhead = KafkaAnalysisUtils.analysisBGCMDHEAD(bGCMDHEADArray);
            bgcmd.setBgCmdHead(bgcmdhead);
            int number = bgcmdhead.getNParamNum();
            System.out.println("number:"+number);
            ArrayList<BGParam> bgParams = new ArrayList<>();
            for (int j = 0; j < number; j++) {
                byte[] beamsBytes = ByteUtils.splitArray(bytes, (i + 1) * 166 + HEAD_SIZE + sumLength, 94);
                BGParam bgParam = KafkaAnalysisUtils.analysisBGParam(beamsBytes);
                bgParams.add(bgParam);
                sumLength += 94;
            }
            bgcmd.setParamArray(bgParams);

            bgcmds.add(bgcmd);
        }

        for(BGCMD bgcmd :bgcmds){
            System.out.println("bgcmd:: " + bgcmd);

            bgcmdService.saveBGCMDInfo(bgcmd);
        }

    }


    /**
     * @description: 去除字符串尾部\0
     * @param: [string]
     * @return: java.lang.String
     * @author YueXQ
     * @date: 2021/11/27 10:29
    **/
    private String filterCode(String string) {
        if (string != null) {
            string = string.trim();
            byte[] zero = new byte[1];
            zero[0] = (byte) 0;
            String s = new String(zero);
            string = string.replaceAll(s, "");
        }
        return string;
    }
}
